package org.ala.tiktools.pacemaker.fixed_interval;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.pacemaker.*;
import org.ala.tiktools.pacemaker.commons.MathUtils;
import org.ala.tiktools.pacemaker.queues.PacemakerQueue;
import org.ala.tiktools.pacemaker.wakeups.FixedIntervalTaskItem;
import org.ala.tiktools.pacemaker.wakeups.WakeupResult;
import org.slf4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 固定间隔的起搏器
 *
 * @author ala
 * @date 2024-12-18 15:27
 */
public class FixedIntervalPacemaker extends AbstractPacemaker {


    static Logger log = LoggerFactory.getLogger("tiktools", "pacemaker");


    /**
     * 起搏器名称
     */
    protected String name;
    /**
     *  间隔毫秒
     *  <p> 默认100ms
     */
    protected long interval = 100L;
    /**
     *  任务队列槽
     */
    protected Map<Integer, PacemakerQueue<FixedIntervalTaskItem>> slots;
    /**
     *  当前的最大公约数，最大unit
     */
    protected volatile int gcd = -1, mxUnit = -1;
    /**
     *  多少次轮训结束后，重置一次最大公约数
     */
    protected int reGcdTimes = 100;
    /**
     *  gcd重置锁
     *  <p> gcd重置会涉及到空槽的清理，是个很重的操作
     */
    protected final Object LOCK_RESET_GCD = new Object();
    /**
     *  起搏线程
     */
    protected PacemakerThread thread;

    public FixedIntervalPacemaker(String name) {
        this(name, 100, 100);
    }
    public FixedIntervalPacemaker(String name, long interval, int reGcdTimes) {
        this.name = name;
        this.interval = interval;
        this.reGcdTimes = reGcdTimes;
        this.slots = new ConcurrentHashMap<>();
    }

    @Override
    public void appendWakeup(long interval, IWakeuper wakeuper, IWakeupError error, IReject reject) {
        //  空任务啥也不做
        if (wakeuper == null) {
            return;
        }
        int unit = (int)interval;

        //  将任务扔进槽内
        appendToQueue(new FixedIntervalTaskItem(unit, wakeuper, error, reject));

        //  更新最大公约数
        updateGcd(unit);
    }
    /**
     *  将任务扔进槽内
     */
    protected void appendToQueue(FixedIntervalTaskItem task) {
        //  将任务扔进槽内
        PacemakerQueue<FixedIntervalTaskItem> queue = slots.get(task.getUnit());
        synchronized (LOCK_RESET_GCD) {
            if (queue == null) {
                //  只取后8位 [-128, 127] 做锁
                synchronized (Integer.valueOf(task.getUnit() & 0xFF)) {
                    if ((queue = slots.get(task.getUnit())) == null) {
                        queue = new PacemakerQueue<>();
                        slots.put(task.getUnit(), queue);
                    }
                }
            }
        }
        queue.offer(task);
    }
    /**
     * 更新最大公约数
     */
    protected void updateGcd(int unit) {
        //  如果最大公约数为空，说明是第一次append任务，开启心跳线程
        if (this.gcd == -1) {
            synchronized (this) {
                if (this.gcd == -1) {
                    this.gcd = this.mxUnit = unit;
                    thread = new PacemakerThread(this);
                    thread.resetGcd(gcd, unit);
                    thread.go();
                    return;
                }
            }
        }

        this.mxUnit = thread.mxCnt = Math.max(this.mxUnit, unit);
        //  更新最大公约数
        int gcd = MathUtils.gcd(this.gcd, unit);
        if (gcd < this.gcd) {
            synchronized (thread) {
                if (gcd < this.gcd) {
                    this.gcd = gcd;
                    //  重置最大公约数
                    thread.resetGcd(gcd, this.mxUnit);
                    //  此时延迟比之前缩小了，所以立即唤醒之后要重新等待新的延迟。按照新的interval等待到下一个整点
                    long now = System.currentTimeMillis();
                    thread.tmpWaitTimestamp = now + (thread.interval - (now - thread.lastWaitTimestamp) % thread.interval);
                    thread.notify();
                }
            }
        }
    }


    /**
     *  心跳
     */
    protected void heart(int cnt) {
        //  分解因子
        int[] fs = MathUtils.factorize(cnt);

        //  每个因子 = 触发任务槽
        for (int f : fs) {
            PacemakerQueue<FixedIntervalTaskItem> queue = slots.get(f);
            AtomicInteger tcnt = new AtomicInteger(0);
            if (queue != null) {
                queue.clearForEach(t -> {
                    doWakeup(t.getTask(), t.getError(), t.getReject(),
                            //  任务成功毁掉（成功拿到返回值）
                            (r) -> {
                                if (WakeupResult.GO_ON.equals(r)) {
                                    appendToQueue(t);
                                }
                            },
                            //  无论成功失败都回调
                            () -> {
                                //  任务计数器，当最后一个任务跑完时，执行finished回调
                                if (tcnt.incrementAndGet() == queue.size()) {
                                    FixedIntervalPacemaker.this.finishedIterator();
                                }
                            });
                });
            }
        }
    }
    protected int crtTimes = 0;
    /**
     *  一轮结束，判断是否需要扩大延迟步长
     *  <p> 有问题，任务是异步执行的，只有任务执行完后才会被放回队列。
     *  <p> 这里拿到的队列为空不能作为判断依据
     */
    protected void finishedIterator() {
        //  对当前所有剩余的槽，做一次最大公约数检测，看看比之前是否大了
        if (reGcdTimes <= 0) { return; }

        if (++crtTimes >= reGcdTimes) {
            crtTimes = 0;

            //  判断是否需要重置
            int gcd = computeGcdBySlots(null)[0];
            if (gcd <= this.gcd) {
                return;
            }

            //  如果发现新的gcd比之前的大了，重置一次线程，清理槽位
            //  清理槽位是个很重的操作，所以会有crtTimes次数，跑了一定次数才会触发
            synchronized (LOCK_RESET_GCD) {
                //  再算一次gcd，如果确实增大了，才会重置。gcd增大说明有小粒度的槽已经空了
                List<Integer> emptySlots = new ArrayList<>();
                int[] gcdAndMx = computeGcdBySlots(emptySlots);
                gcd = gcdAndMx[0];
                int mx = gcdAndMx[1];

                if (gcd > this.gcd) {
                    //  清理空槽
                    for (int k : emptySlots) {
                        slots.remove(k);
                    }

                    //  重置gcd
                    this.gcd = gcd;
                    thread.resetGcd(gcd, mx);
                    //  此时延迟比之前缩小了，所以立即唤醒之后要重新等待新的延迟。按照新的interval等待到下一个整点
                    long now = System.currentTimeMillis();
                    long itvl = (thread.interval - (now - thread.lastWaitTimestamp) % thread.interval);
                    thread.tmpWaitTimestamp = now + itvl;
                    //  特殊的，这里要把cnt加上一遍步长。因为reset会把步长清零，这里算成是第一次跑，步长要加回来
                    thread.cnt = thread.step;
                    synchronized (thread) {
                        thread.notify();
                    }
                }
            }
        }
    }
    /**
     *  根据现有槽位重算一次gcd
     */
    protected int[] computeGcdBySlots(List<Integer> emptySlots) {
        Set<Integer> keys = slots.keySet();
        List<Integer> nums = new ArrayList<>();
        for (int k : keys) {
            if (!slots.containsKey(k)) {
                emptySlots.add(k);
                continue;
            }
            //  有个问题，任务异步执行完后才能知道有没有被放回去。此时拿到的queue有可能是空
            if (!slots.get(k).isEmpty()) {
                nums.add(k);
            } else if (emptySlots != null) {
                emptySlots.add(k);
            }
        }
        return MathUtils.gcdAndMax(nums);
    }


    /**
     *  起搏器线程
     */
    static class PacemakerThread extends Thread {
        /**
         *  执行间隔
         */
        protected long interval = 100;
        protected long tmpInterval = 0;
        protected long lastWaitTimestamp = -1, tmpWaitTimestamp = 0;
        /**
         *  当前gcd，第几次起搏，起搏步长
         */
        protected int cnt = 0, step = -1;
        protected int mxCnt = -1;
        protected volatile boolean running = false;
        protected FixedIntervalPacemaker pacemaker;
        public PacemakerThread(FixedIntervalPacemaker pacemaker) {
            this.pacemaker = pacemaker;
            this.setName(String.format("PacemakerThread_%s", pacemaker.name));
        }
        public void run() {
            synchronized (this) {
                while (running) {
                    try {
                        lastWaitTimestamp = System.currentTimeMillis();
                        cnt += step;

                        block(interval);
                        while (tmpWaitTimestamp > System.currentTimeMillis()) {
                            block(tmpWaitTimestamp - System.currentTimeMillis());
                        }
                        if (!running) {return;}
                        tmpInterval = 0;
                        tmpWaitTimestamp = 0;

                        pacemaker.heart(cnt);
                        if (cnt == mxCnt) {
                            cnt = 0;
//                            pacemaker.finishedIterator();
                        }
                    } catch (InterruptedException e) {
                        log.error(e.getMessage(), e);
                    }
                }
            }
        }
        public void go() {
            if (running) {return;}
            running = true;
            super.start();
        }
        public void end() {
            if (!running) {return;}
            this.running = false;
        }
        protected void block(long interval) throws InterruptedException {
            if (interval <= 0) {
                return;
            }
            //  10ms以内自旋掉
            if (interval < 10) {
                long last = System.currentTimeMillis();
                while (System.currentTimeMillis() - last < interval) {}
            } else {
                //  会在外面拿锁
                this.wait(interval);
            }
        }
        /**
         *  重置最大公约数，最大步数
         */
        public void resetGcd(int step, int mxCnt) {
            this.interval = step * pacemaker.interval;
            this.step = step;
            this.cnt = 0;
            this.mxCnt = mxCnt;
        }
    }
}
